udsink

Documentation

Index

• Variables
• type ApplyUDSinkErr
  • func (e *ApplyUDSinkErr) Error() string
  • func (e *ApplyUDSinkErr) IsInternalErr() bool
  • func (e *ApplyUDSinkErr) IsUserUDSinkErr() bool
• type InternalErr
• type SinkApplier
• type UDSgRPCBasedUDSink
  • func NewUDSgRPCBasedUDSink(client sinkclient.Client) *UDSgRPCBasedUDSink
  • func (u *UDSgRPCBasedUDSink) ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error
  • func (u *UDSgRPCBasedUDSink) CloseConn(ctx context.Context) error
  • func (u *UDSgRPCBasedUDSink) IsHealthy(ctx context.Context) error
  • func (u *UDSgRPCBasedUDSink) WaitUntilReady(ctx context.Context) error
• type UserDefinedSink
  • func NewUserDefinedSink(ctx context.Context, vertexInstance *dfv1.VertexInstance, udsink SinkApplier) (*UserDefinedSink, error)
  • func (s *UserDefinedSink) Close() error
  • func (s *UserDefinedSink) GetName() string
  • func (s *UserDefinedSink) GetPartitionIdx() int32
  • func (s *UserDefinedSink) IsFull() bool
  • func (s *UserDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]isb.Offset, []error)

Variables

var (
	WriteToFallbackErr error = &ApplyUDSinkErr{
		UserUDSinkErr: true,
		Message:       "write to fallback sink",
	}

	UnknownUDSinkErr error = &ApplyUDSinkErr{
		UserUDSinkErr: true,
		Message:       "unknown error in udsink",
	}
	NotFoundErr error = &ApplyUDSinkErr{
		UserUDSinkErr: true,
		Message:       "not found in response",
	}
)

Types

type ApplyUDSinkErr

type ApplyUDSinkErr struct {
	UserUDSinkErr bool
	Message       string
	InternalErr
}

ApplyUDSinkErr represents any UDSink related error

func (*ApplyUDSinkErr) Error

func (e *ApplyUDSinkErr) Error() string

func (*ApplyUDSinkErr) IsInternalErr

func (e *ApplyUDSinkErr) IsInternalErr() bool

IsInternalErr is true if this is a platform issue. This is a blocking error.

func (*ApplyUDSinkErr) IsUserUDSinkErr

func (e *ApplyUDSinkErr) IsUserUDSinkErr() bool

IsUserUDSinkErr is true if the problem is due to the user code in the UDSink.

type InternalErr

type InternalErr struct {
	Flag        bool
	MainCarDown bool
}

InternalErr represents errors internal to the platform

type SinkApplier

type SinkApplier interface {
	ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error
}

SinkApplier applies the sink on the read message and gives back a response. Any UserError will be retried here, while InternalErr can be returned and could be retried by the callee.

type UDSgRPCBasedUDSink

type UDSgRPCBasedUDSink struct {
	// contains filtered or unexported fields
}

UDSgRPCBasedUDSink applies user-defined sink over gRPC (over Unix Domain Socket) client/server where server is the UDSink.

func NewUDSgRPCBasedUDSink

func NewUDSgRPCBasedUDSink(client sinkclient.Client) *UDSgRPCBasedUDSink

NewUDSgRPCBasedUDSink returns UDSgRPCBasedUDSink

func (*UDSgRPCBasedUDSink) ApplySink

func (u *UDSgRPCBasedUDSink) ApplySink(ctx context.Context, requests []*sinkpb.SinkRequest) []error

func (*UDSgRPCBasedUDSink) CloseConn

func (u *UDSgRPCBasedUDSink) CloseConn(ctx context.Context) error

CloseConn closes the gRPC client connection.

func (*UDSgRPCBasedUDSink) IsHealthy

func (u *UDSgRPCBasedUDSink) IsHealthy(ctx context.Context) error

IsHealthy checks if the udsink is healthy.

func (*UDSgRPCBasedUDSink) WaitUntilReady

func (u *UDSgRPCBasedUDSink) WaitUntilReady(ctx context.Context) error

WaitUntilReady waits until the udsink is connected.

type UserDefinedSink

type UserDefinedSink struct {
	// contains filtered or unexported fields
}

func NewUserDefinedSink

func NewUserDefinedSink(ctx context.Context, vertexInstance *dfv1.VertexInstance, udsink SinkApplier) (*UserDefinedSink, error)

NewUserDefinedSink returns genericSink type.

func (*UserDefinedSink) Close

func (s *UserDefinedSink) Close() error

func (*UserDefinedSink) GetName

func (s *UserDefinedSink) GetName() string

func (*UserDefinedSink) GetPartitionIdx

func (s *UserDefinedSink) GetPartitionIdx() int32

GetPartitionIdx returns the partition index. for sink it is always 0.

func (*UserDefinedSink) IsFull

func (s *UserDefinedSink) IsFull() bool

func (*UserDefinedSink) Write

func (s *UserDefinedSink) Write(ctx context.Context, messages []isb.Message) ([]isb.Offset, []error)

Write writes to the UDSink container.