@Test public void testGetSplit() throws Exception { TableMeta meta = CatalogUtil.newTableMeta("KAFKA", new TajoConf()); Map<String, String> option = new java.util.HashMap<String, String>(); option.put(KafkaStorageConstants.KAFKA_TOPIC, TOPIC_NAME); option.put(KafkaStorageConstants.KAFKA_FRAGMENT_SIZE, "10"); meta.setPropertySet(new KeyValueSet(option)); TableDesc td = new TableDesc("test_table", null, meta, null); KafkaTablespace kafkaTablespace = TablespaceManager.getByName("cluster1"); List<Fragment> fragmentList = kafkaTablespace.getSplits("", td, false, null); long totalCount = 0; for (int i = 0; i < fragmentList.size(); i++) { totalCount += fragmentList.get(i).getLength(); } assertEquals(100, totalCount); }
@Override public List<Fragment> getSplits(String inputSourceId, TableDesc table, boolean requireSorted, EvalNode filterCondition) throws IOException, TajoException { String topic = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC); int fragmentSize = Integer.parseInt(table.getMeta().getProperty(KafkaStorageConstants.KAFKA_FRAGMENT_SIZE, KafkaStorageConstants.DEFAULT_FRAGMENT_SIZE)); String partitions = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC_PARTITION, KafkaStorageConstants.DEFAULT_PARTITION); List<PartitionInfo> partitionList; if (partitions.equals(KafkaStorageConstants.DEFAULT_PARTITION)) { partitionList = SimpleConsumerManager.getPartitions(uri, topic); } else { partitionList = new LinkedList<>(); List<PartitionInfo> topicPartitions = SimpleConsumerManager.getPartitions(uri, topic); Map<String, PartitionInfo> topicPartitionsMap = new HashMap<>(); for (PartitionInfo partitionInfo : topicPartitions) { topicPartitionsMap.put(Integer.toString(partitionInfo.partition()), partitionInfo); } for (String partitionId : partitions.split(",")) { partitionList.add(topicPartitionsMap.get(partitionId)); } } List<Fragment> fragments = new ArrayList<Fragment>(); for (PartitionInfo partitionInfo : partitionList) { int partitionId = partitionInfo.partition(); String leaderHost = partitionInfo.leader().host(); long lastOffset; long startOffset; try (SimpleConsumerManager simpleConsumerManager = new SimpleConsumerManager(uri, topic, partitionId)) { lastOffset = simpleConsumerManager.getLatestOffset(); startOffset = simpleConsumerManager.getEarliestOffset(); } long messageSize = lastOffset - startOffset; if (0 == lastOffset || 0 == messageSize) continue; if (messageSize <= fragmentSize) { fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, startOffset, lastOffset, partitionId, leaderHost)); } else { long nextFragmentStartOffset = startOffset; while (nextFragmentStartOffset < lastOffset) { long nextFragmentLastOffset = nextFragmentStartOffset + fragmentSize; long fragmentLstOffset = Math.min(nextFragmentLastOffset, lastOffset); fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, nextFragmentStartOffset, fragmentLstOffset, partitionId, leaderHost)); nextFragmentStartOffset = nextFragmentLastOffset; } } } return fragments; }
KafkaTablespace extends Tablespace { @Override public List<Fragment> getSplits(String inputSourceId, TableDesc table, boolean requireSorted, EvalNode filterCondition) throws IOException, TajoException { String topic = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC); int fragmentSize = Integer.parseInt(table.getMeta().getProperty(KafkaStorageConstants.KAFKA_FRAGMENT_SIZE, KafkaStorageConstants.DEFAULT_FRAGMENT_SIZE)); String partitions = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC_PARTITION, KafkaStorageConstants.DEFAULT_PARTITION); List<PartitionInfo> partitionList; if (partitions.equals(KafkaStorageConstants.DEFAULT_PARTITION)) { partitionList = SimpleConsumerManager.getPartitions(uri, topic); } else { partitionList = new LinkedList<>(); List<PartitionInfo> topicPartitions = SimpleConsumerManager.getPartitions(uri, topic); Map<String, PartitionInfo> topicPartitionsMap = new HashMap<>(); for (PartitionInfo partitionInfo : topicPartitions) { topicPartitionsMap.put(Integer.toString(partitionInfo.partition()), partitionInfo); } for (String partitionId : partitions.split(",")) { partitionList.add(topicPartitionsMap.get(partitionId)); } } List<Fragment> fragments = new ArrayList<Fragment>(); for (PartitionInfo partitionInfo : partitionList) { int partitionId = partitionInfo.partition(); String leaderHost = partitionInfo.leader().host(); long lastOffset; long startOffset; try (SimpleConsumerManager simpleConsumerManager = new SimpleConsumerManager(uri, topic, partitionId)) { lastOffset = simpleConsumerManager.getLatestOffset(); startOffset = simpleConsumerManager.getEarliestOffset(); } long messageSize = lastOffset - startOffset; if (0 == lastOffset || 0 == messageSize) continue; if (messageSize <= fragmentSize) { fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, startOffset, lastOffset, partitionId, leaderHost)); } else { long nextFragmentStartOffset = startOffset; while (nextFragmentStartOffset < lastOffset) { long nextFragmentLastOffset = nextFragmentStartOffset + fragmentSize; long fragmentLstOffset = Math.min(nextFragmentLastOffset, lastOffset); fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, nextFragmentStartOffset, fragmentLstOffset, partitionId, leaderHost)); nextFragmentStartOffset = nextFragmentLastOffset; } } } return fragments; } }
KafkaTablespace extends Tablespace { @Override public List<Fragment> getSplits(String inputSourceId, TableDesc table, boolean requireSorted, EvalNode filterCondition) throws IOException, TajoException { String topic = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC); int fragmentSize = Integer.parseInt(table.getMeta().getProperty(KafkaStorageConstants.KAFKA_FRAGMENT_SIZE, KafkaStorageConstants.DEFAULT_FRAGMENT_SIZE)); String partitions = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC_PARTITION, KafkaStorageConstants.DEFAULT_PARTITION); List<PartitionInfo> partitionList; if (partitions.equals(KafkaStorageConstants.DEFAULT_PARTITION)) { partitionList = SimpleConsumerManager.getPartitions(uri, topic); } else { partitionList = new LinkedList<>(); List<PartitionInfo> topicPartitions = SimpleConsumerManager.getPartitions(uri, topic); Map<String, PartitionInfo> topicPartitionsMap = new HashMap<>(); for (PartitionInfo partitionInfo : topicPartitions) { topicPartitionsMap.put(Integer.toString(partitionInfo.partition()), partitionInfo); } for (String partitionId : partitions.split(",")) { partitionList.add(topicPartitionsMap.get(partitionId)); } } List<Fragment> fragments = new ArrayList<Fragment>(); for (PartitionInfo partitionInfo : partitionList) { int partitionId = partitionInfo.partition(); String leaderHost = partitionInfo.leader().host(); long lastOffset; long startOffset; try (SimpleConsumerManager simpleConsumerManager = new SimpleConsumerManager(uri, topic, partitionId)) { lastOffset = simpleConsumerManager.getLatestOffset(); startOffset = simpleConsumerManager.getEarliestOffset(); } long messageSize = lastOffset - startOffset; if (0 == lastOffset || 0 == messageSize) continue; if (messageSize <= fragmentSize) { fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, startOffset, lastOffset, partitionId, leaderHost)); } else { long nextFragmentStartOffset = startOffset; while (nextFragmentStartOffset < lastOffset) { long nextFragmentLastOffset = nextFragmentStartOffset + fragmentSize; long fragmentLstOffset = Math.min(nextFragmentLastOffset, lastOffset); fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, nextFragmentStartOffset, fragmentLstOffset, partitionId, leaderHost)); nextFragmentStartOffset = nextFragmentLastOffset; } } } return fragments; } KafkaTablespace(String name, URI uri, JSONObject config); }
KafkaTablespace extends Tablespace { @Override public List<Fragment> getSplits(String inputSourceId, TableDesc table, boolean requireSorted, EvalNode filterCondition) throws IOException, TajoException { String topic = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC); int fragmentSize = Integer.parseInt(table.getMeta().getProperty(KafkaStorageConstants.KAFKA_FRAGMENT_SIZE, KafkaStorageConstants.DEFAULT_FRAGMENT_SIZE)); String partitions = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC_PARTITION, KafkaStorageConstants.DEFAULT_PARTITION); List<PartitionInfo> partitionList; if (partitions.equals(KafkaStorageConstants.DEFAULT_PARTITION)) { partitionList = SimpleConsumerManager.getPartitions(uri, topic); } else { partitionList = new LinkedList<>(); List<PartitionInfo> topicPartitions = SimpleConsumerManager.getPartitions(uri, topic); Map<String, PartitionInfo> topicPartitionsMap = new HashMap<>(); for (PartitionInfo partitionInfo : topicPartitions) { topicPartitionsMap.put(Integer.toString(partitionInfo.partition()), partitionInfo); } for (String partitionId : partitions.split(",")) { partitionList.add(topicPartitionsMap.get(partitionId)); } } List<Fragment> fragments = new ArrayList<Fragment>(); for (PartitionInfo partitionInfo : partitionList) { int partitionId = partitionInfo.partition(); String leaderHost = partitionInfo.leader().host(); long lastOffset; long startOffset; try (SimpleConsumerManager simpleConsumerManager = new SimpleConsumerManager(uri, topic, partitionId)) { lastOffset = simpleConsumerManager.getLatestOffset(); startOffset = simpleConsumerManager.getEarliestOffset(); } long messageSize = lastOffset - startOffset; if (0 == lastOffset || 0 == messageSize) continue; if (messageSize <= fragmentSize) { fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, startOffset, lastOffset, partitionId, leaderHost)); } else { long nextFragmentStartOffset = startOffset; while (nextFragmentStartOffset < lastOffset) { long nextFragmentLastOffset = nextFragmentStartOffset + fragmentSize; long fragmentLstOffset = Math.min(nextFragmentLastOffset, lastOffset); fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, nextFragmentStartOffset, fragmentLstOffset, partitionId, leaderHost)); nextFragmentStartOffset = nextFragmentLastOffset; } } } return fragments; } KafkaTablespace(String name, URI uri, JSONObject config); @Override long getTableVolume(TableDesc table, Optional<EvalNode> filter); @Override void close(); @Override void createTable(TableDesc tableDesc, boolean ifNotExists); @Override void purgeTable(TableDesc tableDesc); @Override URI getTableUri(TableMeta meta, String databaseName, String tableName); @Override List<Fragment> getSplits(String inputSourceId,
      TableDesc table,
      boolean requireSorted,
      EvalNode filterCondition); @Override StorageProperty getProperty(); @Override FormatProperty getFormatProperty(TableMeta meta); @Override TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc, Schema inputSchema,
      SortSpec[] sortSpecs, TupleRange dataRange); @Override void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema); @Override void prepareTable(LogicalNode node); @Override Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId, LogicalPlan plan, Schema schema,
      TableDesc tableDesc); @Override void rollbackTable(LogicalNode node); @Override URI getStagingUri(OverridableConf context, String queryId, TableMeta meta); }
KafkaTablespace extends Tablespace { @Override public List<Fragment> getSplits(String inputSourceId, TableDesc table, boolean requireSorted, EvalNode filterCondition) throws IOException, TajoException { String topic = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC); int fragmentSize = Integer.parseInt(table.getMeta().getProperty(KafkaStorageConstants.KAFKA_FRAGMENT_SIZE, KafkaStorageConstants.DEFAULT_FRAGMENT_SIZE)); String partitions = table.getMeta().getProperty(KafkaStorageConstants.KAFKA_TOPIC_PARTITION, KafkaStorageConstants.DEFAULT_PARTITION); List<PartitionInfo> partitionList; if (partitions.equals(KafkaStorageConstants.DEFAULT_PARTITION)) { partitionList = SimpleConsumerManager.getPartitions(uri, topic); } else { partitionList = new LinkedList<>(); List<PartitionInfo> topicPartitions = SimpleConsumerManager.getPartitions(uri, topic); Map<String, PartitionInfo> topicPartitionsMap = new HashMap<>(); for (PartitionInfo partitionInfo : topicPartitions) { topicPartitionsMap.put(Integer.toString(partitionInfo.partition()), partitionInfo); } for (String partitionId : partitions.split(",")) { partitionList.add(topicPartitionsMap.get(partitionId)); } } List<Fragment> fragments = new ArrayList<Fragment>(); for (PartitionInfo partitionInfo : partitionList) { int partitionId = partitionInfo.partition(); String leaderHost = partitionInfo.leader().host(); long lastOffset; long startOffset; try (SimpleConsumerManager simpleConsumerManager = new SimpleConsumerManager(uri, topic, partitionId)) { lastOffset = simpleConsumerManager.getLatestOffset(); startOffset = simpleConsumerManager.getEarliestOffset(); } long messageSize = lastOffset - startOffset; if (0 == lastOffset || 0 == messageSize) continue; if (messageSize <= fragmentSize) { fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, startOffset, lastOffset, partitionId, leaderHost)); } else { long nextFragmentStartOffset = startOffset; while (nextFragmentStartOffset < lastOffset) { long nextFragmentLastOffset = nextFragmentStartOffset + fragmentSize; long fragmentLstOffset = Math.min(nextFragmentLastOffset, lastOffset); fragments.add(new KafkaFragment(table.getUri(), inputSourceId, topic, nextFragmentStartOffset, fragmentLstOffset, partitionId, leaderHost)); nextFragmentStartOffset = nextFragmentLastOffset; } } } return fragments; } KafkaTablespace(String name, URI uri, JSONObject config); @Override long getTableVolume(TableDesc table, Optional<EvalNode> filter); @Override void close(); @Override void createTable(TableDesc tableDesc, boolean ifNotExists); @Override void purgeTable(TableDesc tableDesc); @Override URI getTableUri(TableMeta meta, String databaseName, String tableName); @Override List<Fragment> getSplits(String inputSourceId,
      TableDesc table,
      boolean requireSorted,
      EvalNode filterCondition); @Override StorageProperty getProperty(); @Override FormatProperty getFormatProperty(TableMeta meta); @Override TupleRange[] getInsertSortRanges(OverridableConf queryContext, TableDesc tableDesc, Schema inputSchema,
      SortSpec[] sortSpecs, TupleRange dataRange); @Override void verifySchemaToWrite(TableDesc tableDesc, Schema outSchema); @Override void prepareTable(LogicalNode node); @Override Path commitTable(OverridableConf queryContext, ExecutionBlockId finalEbId, LogicalPlan plan, Schema schema,
      TableDesc tableDesc); @Override void rollbackTable(LogicalNode node); @Override URI getStagingUri(OverridableConf context, String queryId, TableMeta meta); static final StorageProperty KAFKA_STORAGE_PROPERTIES; static final FormatProperty KAFKA_FORMAT_PROPERTIES; }